package com.kingsoft.dc.khaos.module.mpp.operator;

import com.kingsoft.dc.khaos.module.mpp.client.ConfVars;
import com.kingsoft.dc.khaos.module.mpp.client.KhaosContants;
import com.kingsoft.dc.khaos.module.mpp.client.MppClient;
import com.kingsoft.dc.khaos.module.mpp.exception.MppClientException;
import com.kingsoft.dc.khaos.module.mpp.exception.MppSqlException;
import com.kingsoft.dc.khaos.module.mpp.metadata.CombineStrategy;
import com.kingsoft.dc.khaos.module.mpp.metadata.MppConfig;
import com.kingsoft.dc.khaos.module.mpp.model.ColumnEntity;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.SqlBuilderUtils;
import com.kingsoft.dc.khaos.plugin.core.KhaosContext;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

/**
 * 全量切片.
 */
public class TySnapshotAll extends CombineStrategy {

    @Override
    public void execute(KhaosContext kc, String moduleId, MppConfig mppConfig) {
        //初始化MPP client,client已经初始化数据库连接
        MppClient client = null;
        try {
            client = new MppClient(mppConfig, kc);

            //获取操作的库表参数
            //比如获取源表所属的schema
            String srcSchema = String.format("\"%s\"", mppConfig.getSrcSchema());
            log().info("源库名srcSchema===>  " + srcSchema);
            //源表名
            String srcTable = String.format("\"%s\"", mppConfig.getSrcTable());
            log().info("源表名srcTable===>  " + srcTable);

            String filter = mppConfig.getDataFilter();

            //目标表所属schema名
            String tarSchema = String.format("\"%s\"", mppConfig.getTarSchema());
            log().info("目标库名tarSchema===>  " + tarSchema);
            //目标表名
            String tarTable = String.format("\"%s\"", mppConfig.getTarTable());
            log().info("目标表名tarTable===>  " + tarTable);

            //目标表名(上一业务日期)
            String preTable = mppConfig.getPreTable();

            //减档表所属schema名
            String delSchema = String.format("\"%s\"", mppConfig.getDelSchema());
            log().info("减档库名delSchema===>  " + delSchema);
            //减档表名
            String delTable = String.format("\"%s\"", mppConfig.getDelTable());
            log().info("减档表名delTable===>  " + delTable);

            //主键列表
            List<String> keys = mppConfig.getpKeys();

            //获取源档类型
            String src_Type = mppConfig.getSrcType();
            //获取主键字符串
            String pKeys = SqlBuilderUtils.selKeys(tarTable, keys);
            //是否物理删除
            boolean isDeleted = mppConfig.isDeleted();
            //获取追加字段名
            //比如获取[数据日期]字段名
            String bizDateColName = client.getAppendColName(ConfVars.P9_DATA_DATE);

            //获取追加字段值，比如获取[数据日期]字段值
            String bizDate = kc.getConf().getString(KhaosContants.BIZ_DATE);
            String batchNo = kc.getConf().getString(KhaosContants.BATCH_NO);
            String jobName = kc.getConf().getString(KhaosContants.JOB_NAME);
            String jobId = kc.getConf().getString(KhaosContants.JOB_INSTANCE_ID);
            //获取表全字段信息如id,name
            List<String> columnEntities = client.getColumnNameList(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            log().info("源表全字段信息===>  " + StringUtils.join(columnEntities, ","));

            //获取表及表全字段信息
            List<ColumnEntity> tableColumns = client.getColumnInfo(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            //获取目标表主键字段信息
            String columns = SqlBuilderUtils.selKeys(tarTable, columnEntities);
            log().info("目标表主键字段信息===>  " + columns);

            //获取两表关联主键SQL字符串（源表与目标表）
            String pKs = SqlBuilderUtils.joinKeys(srcTable, tarTable, keys);
            log().info("源表与目标表关联主键SQL字符串===>  " + pKs);
            //获取两表关联主键SQL字符串（减档表与目标表）
            String pKsd = SqlBuilderUtils.joinKeys(delTable, tarTable, keys);
            log().info("减档表与目标表关联主键SQL字符串===>  " + pKsd);
            //获取两表关联字段空值处理SQL字符串
            String columnNulls = SqlBuilderUtils.replaceNull(srcTable, tarTable, tableColumns);
            //获取表别名+表全字段信息如s.id,s.name
            String selTabCols = SqlBuilderUtils.selTabCols(srcTable, columnEntities);
            //获取表别名+表全字段信息如s.id,s.name
            String selTabColst = SqlBuilderUtils.selTabCols(tarTable, columnEntities);
            log().info("目标表别名及目标表全字段信息===>  " + selTabColst);
            //获取表别名+表全字段信息如s.id,s.name
            // String selTabColst1 = SqlBuilderUtils.selTabCols(tarTable + "1",columnEntities);

            //支持重跑算子SQL
            String rerun_sql = rerunSql(tarSchema, tarTable, bizDate, batchNo, client);
//            System.out.println(rerun_sql);
            client.execInsert(rerun_sql);

            //拼接全量切片表全量档特殊算子SQL
            String exeu_sql = "";
            if (null == filter || filter.equals("")) {
                exeu_sql = buildSqlall(src_Type, tarSchema, tarTable, srcSchema, srcTable, bizDate, jobName, batchNo, columns, client);
            } else {
                exeu_sql = buildSqlallWithFilter(src_Type, tarSchema, tarTable, srcSchema, srcTable, bizDate, jobName, batchNo, columns, client, filter);
            }
            //            System.out.println(exeu_sql);
            client.execInsert(exeu_sql);
            String exe_sql = "";
            //拼接全量切片表算子SQL
            if (null == filter || filter.equals("")) {
                exe_sql = buildSql(src_Type, tarSchema, tarTable, srcSchema, srcTable, delSchema, delTable, bizDate, jobName, batchNo, jobId, columns, pKs, pKsd, keys, selTabColst, isDeleted, client);
            } else {
                exe_sql = buildSqlWithFilter(src_Type, tarSchema, tarTable, srcSchema, srcTable, delSchema, delTable, bizDate, jobName, batchNo, jobId, columns, pKs, pKsd, keys, selTabColst, isDeleted, client, filter);
            }
            client.execInsert(exe_sql);
        } catch (MppClientException e) {
            e.printStackTrace();
            throw new MppSqlException("执行MPP SQL 异常");
        } finally {
            if (client != null) {
                client.close();
            }
        }
    }

    //重跑作业清理
    public String rerunSql(String tarSchema, String tarTable, String bizDate, String batchNo, MppClient client) {
        System.out.println("TySnapshotAll build rerunSql start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        //重跑作业清理
        String del_sql = String.format("\n BEGIN WORK; \n" +
                "delete from %s.%s \n" +
                "        where " + tyDataDate + " = to_date('%s','YYYYMMDD')\n" +
                "          and " + tyBatchNumber + " = '%s';\n", tarSchema, tarTable, bizDate, batchNo);
        System.out.println("TySnapshotAll build rerunSql end...");
        return del_sql;
    }

    //拼接全量切片表全量档算子SQL实现类
    public String buildSqlall(String src_Type, String tarSchema, String tarTable, String srcSchema, String srcTable, String bizDate, String jobName, String batchNo, String columns, MppClient client) {
        System.out.println("TySnapshotAll buildSqlall start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tySrcFlag = client.getAppendColName(ConfVars.P9_SRC_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        //全量处理逻辑，不能判断是否物理删除
        if (src_Type.equalsIgnoreCase("ALL")) {
            all_sql = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tySrcFlag + ",\n" +
                    "\t      " + tyJobName + "\t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                    "\t      '%s' as " + tyBatchNumber + ",\n" +
                    "\t      '0' as " + tySrcFlag + ",\n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t\tfrom %s.%s;\n", tarSchema, tarTable, columns, columns, bizDate, batchNo, jobName, srcSchema, srcTable);
        }
        System.out.println("TySnapshotAll buildSqlall end...");
        return all_sql;
    }

    //拼接全量切片表全量档算子SQL实现类
    public String buildSqlallWithFilter(String src_Type,
                                        String tarSchema,
                                        String tarTable,
                                        String srcSchema,
                                        String srcTable,
                                        String bizDate,
                                        String jobName,
                                        String batchNo,
                                        String columns,
                                        MppClient client,
                                        String filter) {
        System.out.println("TySnapshotAll buildSqlallWithFilter start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tySrcFlag = client.getAppendColName(ConfVars.P9_SRC_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        String schemaTblFilter = String.format("(select * from %s.%s where %s)", srcSchema, srcTable, filter);
        //全量处理逻辑，不能判断是否物理删除
        if (src_Type.equalsIgnoreCase("ALL")) {
            all_sql = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tySrcFlag + ",\n" +
                    "\t      " + tyJobName + "\t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                    "\t      '%s' as " + tyBatchNumber + ",\n" +
                    "\t      '0' as " + tySrcFlag + ",\n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t\tfrom %s %s;\n", tarSchema, tarTable, columns, columns, bizDate, batchNo, jobName, schemaTblFilter, srcTable);
        }
        System.out.println("TySnapshotAll buildSqlallWithFilter end...");
        return all_sql;
    }

    //拼接全量切片表增量减量档算子SQL实现类
    public String buildSql(String src_Type,
                           String tarSchema,
                           String tarTable,
                           String srcSchema,
                           String srcTable,
                           String delSchema,
                           String delTable,
                           String bizDate,
                           String jobName,
                           String batchNo,
                           String jobId,
                           String columns,
                           String pKs,
                           String pKsd,
                           List<String> keys,
                           String selTabColst,
                           boolean isDeleted,
                           MppClient client) {
        System.out.println("TySnapshotAll buildSql start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tySrcFlag = client.getAppendColName(ConfVars.P9_SRC_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        if ((src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            //增量且有物理删除
            if (src_Type.equalsIgnoreCase("ADD") && isDeleted) {
                //增量且有物理删除
                String exe_sql_one = String.format("insert into %s.%s (\n" +
                                "\t      %s,\n" +
                                "\t      " + tyDataDate + ",\n" +
                                "\t      " + tyBatchNumber + ",\n" +
                                "\t      " + tySrcFlag + ",\n" +
                                "\t      " + tyJobName + " \n" +
                                "\t) select \n" +
                                "\t      %s,\n" +
                                "\t      " + tyDataDate + ",\n" +
                                "\t      " + tyBatchNumber + ",\n" +
                                "\t      " + tySrcFlag + ",\n" +
                                "\t      " + tyJobName + " \n" +
                                "\t  from (\n" +
                                "\t    select \n" +
                                "\t       %s,\n" +
                                "\t       to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                                "\t       '%s' as " + tyBatchNumber + ",\n" +
                                "\t       '0' as " + tySrcFlag + ",  \n" +
                                "\t       '%s' as " + tyJobName + "\n" +
                                "\t    from  %s.%s \n" +
                                "\t    where " + tyDataDate + " = ( select max(" + tyDataDate + ") from %s.%s ) \n" +
                                "\t ) %s\n" +
                                "\t left join %s.%s %s   \n" +
                                "\t on %s\n" +
                                "\t where %s.%s is null; \n", tarSchema, tarTable, columns, selTabColst, columns, bizDate, batchNo, jobName, tarSchema, tarTable, tarSchema
                        , tarTable, tarTable, delSchema, delTable, delTable, pKsd, delTable, keys.get(0));
                all_sql = all_sql + exe_sql_one + "\n COMMIT ; \n ";
                ;
            }

            //1.增量或是2.增量且有物理删除，注：DEL表示增量有物理删除，这时候有供减量源表
            //创建动态临时表
            String createvt_sql = String.format("drop table if exists %s.tmp_%s; \n", tarSchema, jobId)
                    + String.format("create table %s.tmp_%s as select * from  %s.%s limit 0; \n", tarSchema, jobId, tarSchema, tarTable)
                    + String.format("truncate table %s.tmp_%s; \n", tarSchema, jobId);

            //插入临时数据
            String exe_sql_a_d_one = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tySrcFlag + ",\n" +
                            "\t      " + tyJobName + " \n" +
                            "\t) select \n" +
                            "\t       %s,\n" +
                            "\t       to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                            "\t       '%s' as " + tyBatchNumber + ",\n" +
                            "\t       '1' as " + tySrcFlag + ",  \n" +
                            "\t       '%s' as " + tyJobName + "\n" +
                            "\t\tFROM %s.%s \n" +
                            "\t\tunion all \n" +
                            "\t\t select \n" +
                            "\t       %s,\n" +
                            "\t       to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                            "\t       '%s' as " + tyBatchNumber + ",\n" +
                            "\t       '0' as " + tySrcFlag + ",  \n" +
                            "\t       '%s' as " + tyJobName + "\n" +
                            "\t\t  from (select * from %s.%s where " + tyDataDate + " = ( select max(" + tyDataDate + ") from %s.%s ) ) %s\n" +
                            "\t\t  left join %s.%s %s\n" +
                            "\t\t  on %s \n" +
                            "\t\t  where %s.%s is null;\n", tarSchema, jobId, columns, columns, bizDate, batchNo, jobName, srcSchema, srcTable, selTabColst, bizDate, batchNo, jobName, tarSchema, tarTable, tarSchema, tarTable, tarTable
                    , srcSchema, srcTable, srcTable, pKs, srcTable, keys.get(0));
            //这里是删除处理减量导致多出的记录
            String exe_sql_a_d_two = String.format("delete from %s.%s where " + tyDataDate + " = to_date('%s','YYYYMMDD') \n" +
                    "\t        and " + tyBatchNumber + " = '%s';\n", tarSchema, tarTable, bizDate, batchNo);
            //插入目标表
            String exe_sql_a_d_three = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tySrcFlag + ",\n" +
                    "\t      " + tyJobName + " \n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tySrcFlag + ",\n" +
                    "\t      " + tyJobName + " \n" +
                    "\t\tfrom %s.tmp_%s;\n", tarSchema, tarTable, columns, columns, tarSchema, jobId);

            //最后删除tmp表
            String exe_sql_a_d_four = String.format("drop table if exists  %s.tmp_%s ;\n", tarSchema, jobId);

            all_sql = all_sql + createvt_sql + exe_sql_a_d_one + exe_sql_a_d_two + exe_sql_a_d_three + exe_sql_a_d_four;
        }
        System.out.println("TySnapshotAll buildSql end...");
        return all_sql + "\n COMMIT ; \n ";
    }

    //拼接全量切片表增量减量档算子SQL实现类
    public String buildSqlWithFilter(String src_Type,
                                     String tarSchema,
                                     String tarTable,
                                     String srcSchema,
                                     String srcTable,
                                     String delSchema,
                                     String delTable,
                                     String bizDate,
                                     String jobName,
                                     String batchNo,
                                     String jobId,
                                     String columns,
                                     String pKs,
                                     String pKsd,
                                     List<String> keys,
                                     String selTabColst,
                                     boolean isDeleted,
                                     MppClient client,
                                     String filter) {
        System.out.println("TySnapshotAll buildSqlWithFilter start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tySrcFlag = client.getAppendColName(ConfVars.P9_SRC_FLAG);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        String schemaTblFilter = String.format("(select * from %s.%s where %s)", srcSchema, srcTable, filter);
        String alis = "f";
        if ((src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            //增量且有物理删除
            if (src_Type.equalsIgnoreCase("ADD") && isDeleted) {
                //增量且有物理删除
                String exe_sql_one = String.format("insert into %s.%s (\n" +
                                "\t      %s,\n" +
                                "\t      " + tyDataDate + ",\n" +
                                "\t      " + tyBatchNumber + ",\n" +
                                "\t      " + tySrcFlag + ",\n" +
                                "\t      " + tyJobName + " \n" +
                                "\t) select \n" +
                                "\t      %s,\n" +
                                "\t      " + tyDataDate + ",\n" +
                                "\t      " + tyBatchNumber + ",\n" +
                                "\t      " + tySrcFlag + ",\n" +
                                "\t      " + tyJobName + " \n" +
                                "\t  from (\n" +
                                "\t    select \n" +
                                "\t       %s,\n" +
                                "\t       to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                                "\t       '%s' as " + tyBatchNumber + ",\n" +
                                "\t       '0' as " + tySrcFlag + ",  \n" +
                                "\t       '%s' as " + tyJobName + "\n" +
                                "\t    from  %s.%s \n" +
                                "\t    where " + tyDataDate + " = ( select max(" + tyDataDate + ") from %s.%s ) \n" +
                                "\t ) %s\n" +
                                "\t left join %s.%s %s   \n" +
                                "\t on %s\n" +
                                "\t where %s.%s is null; \n", tarSchema, tarTable, columns, selTabColst, columns, bizDate, batchNo, jobName, tarSchema, tarTable, tarSchema
                        , tarTable, tarTable, delSchema, delTable, delTable, pKsd, delTable, keys.get(0));
                all_sql = all_sql + exe_sql_one + "\n COMMIT ; \n ";
                ;
            }

            //1.增量或是2.增量且有物理删除，注：DEL表示增量有物理删除，这时候有供减量源表
            //创建动态临时表
            String createvt_sql = String.format("drop table if exists %s.tmp_%s; \n", tarSchema, jobId)
                    + String.format("create table %s.tmp_%s as select * from  %s.%s limit 0; \n", tarSchema, jobId, tarSchema, tarTable)
                    + String.format("truncate table %s.tmp_%s; \n", tarSchema, jobId);

            //插入临时数据
            String exe_sql_a_d_one = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tySrcFlag + ",\n" +
                            "\t      " + tyJobName + " \n" +
                            "\t) select \n" +
                            "\t       %s,\n" +
                            "\t       to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                            "\t       '%s' as " + tyBatchNumber + ",\n" +
                            "\t       '1' as " + tySrcFlag + ",  \n" +
                            "\t       '%s' as " + tyJobName + "\n" +
                            "\t\tFROM %s %s\n" +
                            "\t\tunion all \n" +
                            "\t\t select \n" +
                            "\t       %s,\n" +
                            "\t       to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                            "\t       '%s' as " + tyBatchNumber + ",\n" +
                            "\t       '0' as " + tySrcFlag + ",  \n" +
                            "\t       '%s' as " + tyJobName + "\n" +
                            "\t\t  from (select * from %s.%s where " + tyDataDate + " = ( select max(" + tyDataDate + ") from %s.%s ) ) %s\n" +
                            "\t\t  left join %s %s\n" +
                            "\t\t  on %s \n" +
                            "\t\t  where %s.%s is null;\n", tarSchema, jobId, columns, columns, bizDate, batchNo, jobName, schemaTblFilter, alis, selTabColst, bizDate, batchNo, jobName, tarSchema, tarTable, tarSchema, tarTable, tarTable
                    , schemaTblFilter, srcTable, pKs, srcTable, keys.get(0));
            //这里是删除处理减量导致多出的记录
            String exe_sql_a_d_two = String.format("delete from %s.%s where " + tyDataDate + " = to_date('%s','YYYYMMDD') \n" +
                    "\t        and " + tyBatchNumber + " = '%s';\n", tarSchema, tarTable, bizDate, batchNo);
            //插入目标表
            String exe_sql_a_d_three = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tySrcFlag + ",\n" +
                    "\t      " + tyJobName + " \n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tySrcFlag + ",\n" +
                    "\t      " + tyJobName + " \n" +
                    "\t\tfrom %s.tmp_%s;\n", tarSchema, tarTable, columns, columns, tarSchema, jobId);

            //最后删除tmp表
            String exe_sql_a_d_four = String.format("drop table if exists  %s.tmp_%s ;\n", tarSchema, jobId);

            all_sql = all_sql + createvt_sql + exe_sql_a_d_one + exe_sql_a_d_two + exe_sql_a_d_three + exe_sql_a_d_four;
        }
        System.out.println("TySnapshotAll buildSqlWithFilter end...");
        return all_sql + "\n COMMIT ; \n ";
    }
}
